package gd1;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import util.KafkaUtil;

/**
 * 数据分流处理类
 * 功能：从Kafka的ODS层读取数据，根据表名分流到不同的DWD层Kafka主题
 */
public class Dwd_SideOutPut {
    public static void main(String[] args) throws Exception {
        // 1. 创建Flink流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1); // 设置并行度为1（生产环境应根据实际情况调整）

        // 2. 从Kafka读取ODS层数据
        DataStreamSource<String> ods_shop_ds = KafkaUtil.getKafkaSource(env, "ods_day0906");

        // 3. 执行侧输出分流操作
        SideOutput(ods_shop_ds);

        // 4. 启动作业
        env.execute("DWD_Data_Split_Job"); // 作业名称
    }

    /**
     * 侧输出分流方法
     * @param ods_app_user_shop 原始ODS层数据流
     */
    private static void SideOutput(DataStreamSource<String> ods_app_user_shop) {
        // 1. 定义三个侧输出标签（用于标识不同类型的数据）
        // 1.1 订单数据标签
        OutputTag<String> order_main = new OutputTag<String>("orders") {};
        // 1.2 支付数据标签
        OutputTag<String> payment_detail = new OutputTag<String>("payment") {};
        // 1.3 用户行为数据标签
//        OutputTag<String> user_behavior_log = new OutputTag<String>("user_behavior_log") {};

        // 2. 处理主数据流
        SingleOutputStreamOperator<Object> process = ods_app_user_shop.process(
                new ProcessFunction<String, Object>() {
                    @Override
                    public void processElement(String s, Context context, Collector<Object> collector) throws Exception {
                        // 2.1 解析JSON数据
                        JSONObject object = JSON.parseObject(s);

                        // 2.2 获取表名（从source字段中）
                        String table = object.getJSONObject("source").getString("table");

                        // 2.3 根据表名将数据分流到不同的侧输出
                        if (table.equals("orders")) {
                            // 订单数据输出到orders侧输出流
                            context.output(order_main, object.getString("after"));
                        }
                        if (table.equals("payment")) {
                            // 支付数据输出到payment侧输出流
                            context.output(payment_detail, object.getString("after"));
                        }
//                        if (table.equals("user_behavior_log")) {
//                            // 用户行为数据输出到user_behavior_log侧输出流
//                            context.output(user_behavior_log, object.getString("after"));
//                        }
                    }
                }
        );

        // 3. 获取侧输出流并写入对应的Kafka主题
        // 3.1 订单数据写入dwd_order_09_06主题
        process.getSideOutput(order_main).addSink(KafkaUtil.getKafkaSink("dwd_order_09_06"));
        // 3.2 支付数据写入dwd_payment_09_06主题
        process.getSideOutput(payment_detail).addSink(KafkaUtil.getKafkaSink("dwd_payment_09_06"));
        // 3.3 用户行为数据写入dwd_user_09_06主题
//        process.getSideOutput(user_behavior_log).addSink(KafkaUtil.getKafkaSink("dwd_user_09_06"));
    }
}